package com.isyscore.os.etl.model.dto;

import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.annotation.JsonAlias;
import lombok.Data;

import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @author : zhan9yn
 * @version : 1.0
 * @date : 2022/1/14 3:14 下午
 */
@Data
public class ExceptionDTO {
    @JsonAlias("root-exception")
    private String rootException;

    private Long timestamp;

    @JsonAlias("all-exceptions")
    private List<Map<String,Object>> allExceptions;

    private boolean truncated;

    public static void main(String[] args) {
        String str = "{\n" +
                "    \"root-exception\": \"org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy\\n\\tat org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:118)\\n\\tat org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:80)\\n\\tat org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:233)\\n\\tat org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:224)\\n\\tat org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:215)\\n\\tat org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:669)\\n\\tat org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:89)\\n\\tat org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:447)\\n\\tat sun.reflect.GeneratedMethodAccessor154.invoke(Unknown Source)\\n\\tat sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\\n\\tat java.lang.reflect.Method.invoke(Method.java:498)\\n\\tat org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:305)\\n\\tat org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:212)\\n\\tat org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:77)\\n\\tat org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:158)\\n\\tat akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26)\\n\\tat akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21)\\n\\tat scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123)\\n\\tat akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21)\\n\\tat scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170)\\n\\tat scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)\\n\\tat scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)\\n\\tat akka.actor.Actor$class.aroundReceive(Actor.scala:517)\\n\\tat akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:225)\\n\\tat akka.actor.ActorCell.receiveMessage(ActorCell.scala:592)\\n\\tat akka.actor.ActorCell.invoke(ActorCell.scala:561)\\n\\tat akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258)\\n\\tat akka.dispatch.Mailbox.run(Mailbox.scala:225)\\n\\tat akka.dispatch.Mailbox.exec(Mailbox.scala:235)\\n\\tat akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)\\n\\tat akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)\\n\\tat akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)\\n\\tat akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)\\nCaused by: java.lang.RuntimeException: java.sql.SQLException: No suitable driver found for jdbc:sqlserver://10.30.30.24:1433;databaseName=add;useLOBs=false\\n\\tat com.dtstack.flinkx.util.RetryUtil$Retry.doRetry(RetryUtil.java:137)\\n\\tat com.dtstack.flinkx.util.RetryUtil.executeWithRetry(RetryUtil.java:56)\\n\\tat com.dtstack.flinkx.connector.jdbc.util.JdbcUtil.getConnection(JdbcUtil.java:112)\\n\\tat com.dtstack.flinkx.connector.jdbc.sink.JdbcOutputFormat.getConnection(JdbcOutputFormat.java:366)\\n\\tat com.dtstack.flinkx.connector.jdbc.sink.JdbcOutputFormat.openInternal(JdbcOutputFormat.java:84)\\n\\tat com.dtstack.flinkx.connector.sqlserver.sink.SqlserverOutputFormat.openInternal(SqlserverOutputFormat.java:39)\\n\\tat com.dtstack.flinkx.sink.format.BaseRichOutputFormat.open(BaseRichOutputFormat.java:216)\\n\\tat com.dtstack.flinkx.sink.DtOutputFormatSinkFunction.open(DtOutputFormatSinkFunction.java:95)\\n\\tat org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:34)\\n\\tat org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:102)\\n\\tat org.apache.flink.table.runtime.operators.sink.SinkOperator.open(SinkOperator.java:63)\\n\\tat org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:428)\\n\\tat org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$beforeInvoke$2(StreamTask.java:543)\\n\\tat org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:93)\\n\\tat org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:533)\\n\\tat org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:573)\\n\\tat org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:755)\\n\\tat org.apache.flink.runtime.taskmanager.Task.run(Task.java:570)\\n\\tat java.lang.Thread.run(Thread.java:748)\\nCaused by: java.sql.SQLException: No suitable driver found for jdbc:sqlserver://10.30.30.24:1433;databaseName=add;useLOBs=false\\n\\tat java.sql.DriverManager.getConnection(DriverManager.java:689)\\n\\tat java.sql.DriverManager.getConnection(DriverManager.java:208)\\n\\tat com.dtstack.flinkx.connector.jdbc.util.JdbcUtil.lambda$getConnection$0(JdbcUtil.java:113)\\n\\tat com.dtstack.flinkx.util.RetryUtil$Retry.call(RetryUtil.java:141)\\n\\tat com.dtstack.flinkx.util.RetryUtil$Retry.doRetry(RetryUtil.java:80)\\n\\t... 18 more\\n\",\n" +
                "    \"timestamp\": 1642140330800,\n" +
                "    \"all-exceptions\": [\n" +
                "        {\n" +
                "            \"exception\": \"java.lang.RuntimeException: java.sql.SQLException: No suitable driver found for jdbc:sqlserver://10.30.30.24:1433;databaseName=add;useLOBs=false\\n\\tat com.dtstack.flinkx.util.RetryUtil$Retry.doRetry(RetryUtil.java:137)\\n\\tat com.dtstack.flinkx.util.RetryUtil.executeWithRetry(RetryUtil.java:56)\\n\\tat com.dtstack.flinkx.connector.jdbc.util.JdbcUtil.getConnection(JdbcUtil.java:112)\\n\\tat com.dtstack.flinkx.connector.jdbc.sink.JdbcOutputFormat.getConnection(JdbcOutputFormat.java:366)\\n\\tat com.dtstack.flinkx.connector.jdbc.sink.JdbcOutputFormat.openInternal(JdbcOutputFormat.java:84)\\n\\tat com.dtstack.flinkx.connector.sqlserver.sink.SqlserverOutputFormat.openInternal(SqlserverOutputFormat.java:39)\\n\\tat com.dtstack.flinkx.sink.format.BaseRichOutputFormat.open(BaseRichOutputFormat.java:216)\\n\\tat com.dtstack.flinkx.sink.DtOutputFormatSinkFunction.open(DtOutputFormatSinkFunction.java:95)\\n\\tat org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:34)\\n\\tat org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:102)\\n\\tat org.apache.flink.table.runtime.operators.sink.SinkOperator.open(SinkOperator.java:63)\\n\\tat org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:428)\\n\\tat org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$beforeInvoke$2(StreamTask.java:543)\\n\\tat org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:93)\\n\\tat org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:533)\\n\\tat org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:573)\\n\\tat org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:755)\\n\\tat org.apache.flink.runtime.taskmanager.Task.run(Task.java:570)\\n\\tat java.lang.Thread.run(Thread.java:748)\\nCaused by: java.sql.SQLException: No suitable driver found for jdbc:sqlserver://10.30.30.24:1433;databaseName=add;useLOBs=false\\n\\tat java.sql.DriverManager.getConnection(DriverManager.java:689)\\n\\tat java.sql.DriverManager.getConnection(DriverManager.java:208)\\n\\tat com.dtstack.flinkx.connector.jdbc.util.JdbcUtil.lambda$getConnection$0(JdbcUtil.java:113)\\n\\tat com.dtstack.flinkx.util.RetryUtil$Retry.call(RetryUtil.java:141)\\n\\tat com.dtstack.flinkx.util.RetryUtil$Retry.doRetry(RetryUtil.java:80)\\n\\t... 18 more\\n\",\n" +
                "            \"task\": \"Source: TableSourceScan(table=[[default_catalog, default_database, source0]], fields=[id, name, age, sex]) -> Sink: Sink(table=[default_catalog.default_database.source1], fields=[id, name, age, sex]) (1/1)\",\n" +
                "            \"location\": \"10.244.0.154:37409\",\n" +
                "            \"timestamp\": 1642140330793\n" +
                "        }\n" +
                "    ],\n" +
                "    \"truncated\": false\n" +
                "}";
        ExceptionDTO e = JSONObject.parseObject(str, ExceptionDTO.class);
        str = e.getAllExceptions().parallelStream().map(m -> (String)m.get("exception")).collect(Collectors.joining("\n"));
        System.out.println(str);
        return ;
    }
}
